package service.impl;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import kafka.javaapi.consumer.ConsumerConnector;
import play.Play;
import service.MessagesQueue;
import utils.LogUtil;
import bp.DataSynBP;

public class SchedulePriceDelMessageQueue implements MessagesQueue {

	private  Queue<String> queue = new LinkedList<String>();
	private Integer consumethreshold=Integer.parseInt(Play.configuration.getProperty("SYNC_CONSUMER_BATCH_SIZE", "1000"));
	private static ConsumerConnector consumer;
	private static boolean canconsumer = true;
	public static void setConsumer(ConsumerConnector consumer) {
		SchedulePriceDelMessageQueue.consumer = consumer;
	}

	private static SchedulePriceDelMessageQueue pricedelmessagequeue;
	
	public static SchedulePriceDelMessageQueue getInstance(){
		synchronized(SchedulePriceDelMessageQueue.class){
			if(pricedelmessagequeue==null){
				pricedelmessagequeue = new SchedulePriceDelMessageQueue();
//				pricedelmessagequeue.consumethreshold = SystemParam.SYNC_CONSUMER_BATCH_SIZE;
			}
		}
		return pricedelmessagequeue;
	}

	@Override
	public void addMessage(String message) throws Exception {
		synchronized(queue){
			queue.offer(message);
		}
		checkqueuesize();
	}
	public static boolean canwork(){
		return canconsumer;
	}
	@Override
	public List<String> getMessages(int size) throws Exception {
		// TODO Auto-generated method stub
		return null;
	}
	
	
	
	private SchedulePriceDelMessageQueue(Integer consumethreshold) {
		super();
		this.consumethreshold = consumethreshold;
	}
	
	private SchedulePriceDelMessageQueue(){
		super();
	}
	
	private int checkqueuesize() throws Exception{
		try {
			if(queue.size()>0 && (queue.size()%consumethreshold==0)){
				consumeMessages(1);
			}
		} catch (Exception e) {
			LogUtil.logerror("consumeMessages（）消费批量消息失败", e);
		}
		return 0;
	}

	public int consumeMessages(int type) throws Exception{
		synchronized(queue){
			canconsumer = false;
			synchronized(SchedulePriceMessageQueue.lock){
				try {
					//queue.clear();
					//int index = consumethreshold;
					String message;
					List<String> messages = new ArrayList<String>();
					while((message=queue.poll())!=null){
						messages.add(message);
						//index--;
					}
					if(messages.size()>0){
						DataSynBP datasynbp = new DataSynBP();
						datasynbp.pricedelsyn(messages);
						consumer.commitOffsets();
					}
					
				} catch (Exception e) {
					LogUtil.logerror("consumeMessages（）消费批量消息失败", e);
				}finally{
					canconsumer = true;
				}
			}
		}
		return 0;
	}

}
